package org.budo.canal.server;

import org.budo.canal.zookeeper.running.AbstractServerRunningListener;
import org.budo.support.lang.util.ProcessUtil;
import org.budo.support.spring.util.SpringUtil;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * 调用 start 方法启动后，会初始化并扫描所有 CanalInstance 并启动监听
 * 
 * @author limingwei
 */
@Getter
@Setter
@Slf4j
public abstract class AbstractBudoCanalServer implements ApplicationContextAware, BudoCanalServer, BeanNameAware {
    private CanalInstanceGenerator canalInstanceGenerator;

    private ApplicationContext applicationContext;

    private CanalServerWithEmbedded canalServer;

    private ZkClientx zkClient;

    private String beanName;

    protected abstract void onServerRunningMonitorProcessActiveEnter(CanalServerWithEmbedded canalServer, String destination);

    private ZkClientx zkClientx() {
        if (null != this.getZkClient()) {
            return this.getZkClient();
        }

        return SpringUtil.getBean(applicationContext, ZkClientx.class);
    }

    protected CanalServerWithEmbedded canalServer() {
        if (null != this.getCanalServer()) {
            return this.getCanalServer();
        }

        synchronized (BudoCanalServer.class) {
            if (null != this.getCanalServer()) {
                return this.getCanalServer();
            }

            CanalServerWithEmbedded canalServer = this.initCanalServerWithEmbedded();
            this.setCanalServer(canalServer);
            return canalServer;
        }
    }

    protected CanalServerWithEmbedded initCanalServerWithEmbedded() {
        final CanalServerWithEmbedded canalServer = CanalServerWithEmbedded.instance();
        canalServer.setCanalInstanceGenerator(this.canalInstanceGenerator());

        log.info("#71 before canalServer.start, canalServer=" + canalServer);
        canalServer.start();

        ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
            public ServerRunningMonitor apply(final String destination) {
                ServerRunningMonitor runningMonitor = AbstractBudoCanalServer.this.newServerRunningMonitor(canalServer, destination);
                runningMonitor.init();
                return runningMonitor;
            }
        }));

        return canalServer;
    }

    private ServerRunningMonitor newServerRunningMonitor(final CanalServerWithEmbedded canalServer, final String destination) {
        ServerRunningMonitor runningMonitor = new ServerRunningMonitor();
        runningMonitor.setDestination(destination);

        if (null == this.zkClientx()) {
            log.warn("#81 zkClient is null, this=" + this);
        } else {
            runningMonitor.setZkClient(this.zkClientx());
            log.info("#84 zkClient = " + this.zkClientx());
        }

        Integer currentProcessId = ProcessUtil.getCurrentProcessId();
        long cid = currentProcessId.longValue();

        // beanName 与数据库连接信息相关, 以此为单位进行互斥
        ServerRunningData serverData = new ServerRunningData(cid, this.getBeanName());

        runningMonitor.setServerData(serverData);
        log.info("#90 runningMonitor = " + runningMonitor);

        runningMonitor.setListener(new AbstractServerRunningListener() {
            public void processActiveEnter() {
                AbstractBudoCanalServer.this.onServerRunningMonitorProcessActiveEnter(canalServer, destination);
            }

            public void processActiveExit() {
                log.info("#98 processActiveExit, stop destination " + destination);

                canalServer.stop(destination);
            }
        });
        return runningMonitor;
    }

    private CanalInstanceGenerator canalInstanceGenerator() {
        if (null != this.getCanalInstanceGenerator()) {
            return this.getCanalInstanceGenerator();
        }

        SpringCanalInstanceGenerator canalInstanceGenerator = new SpringCanalInstanceGenerator();
        canalInstanceGenerator.setBeanFactory(this.getApplicationContext());
        return canalInstanceGenerator;
    }
}